Swarm 框架简易入门指南
#AI
什么是 Swarm?
Swarm 是一个简单但功能强大的 Python 框架,它让多个智能助手(称为 Agent)能够协同工作并相互交流。想象一下,这就像一群小蜜蜂一起完成不同任务,每只蜜蜂都有特定技能,它们可以把工作传递给其他更擅长该任务的蜜蜂。
🔔 重要提示:Swarm 现已被 OpenAI Agents SDK 替代,后者是更新更完善的版本。本指南帮助你理解 Swarm 的基本概念,这些概念与新的 Agents SDK 类似。
核心概念(基础构建块)
1. Agent(智能助手)
Agent 就像一个聪明的小助手,它知道如何完成特定任务。每个 Agent 都有:
- 名字(例如"天气助手"或"数学助手")
- 指令(告诉 Agent 应该做什么和如何交谈)
- 函数(Agent 可以使用的特殊工具)
案例1:创建一个简单的助手
from swarm import Swarm, Agent
simple_agent = Agent(
name="简单助手",
instructions="你是一个乐于助人的助手,总是用友善的语气回答问题。",
)案例2:创建一个天气助手
weather_agent = Agent(
name="天气助手",
instructions="你是专门提供天气信息的助手,会告诉用户任何地点的天气。",
functions=[get_weather, send_email]
)案例3:创建一个翻译助手
translator_agent = Agent(
name="翻译助手",
instructions="你是语言翻译专家,可以将任何语言翻译成用户想要的语言。",
functions=[translate_text]
)2. 函数(Functions)
函数就像是 Agent 可以使用的特殊工具。它们让 Agent 能够执行实际操作,比如:
- 查询天气信息
- 发送电子邮件
- 翻译文本
案例1:天气查询函数
def get_weather(location):
"""获取指定地点的天气。地点必须是一个城市名。"""
# 这里会有获取天气的实际代码
return f"{location} 今天天气晴朗,气温25°C。"案例2:发送电子邮件函数
def send_email(recipient, subject, body):
"""发送一封电子邮件。"""
print(f"正在发送邮件...")
print(f"收件人: {recipient}")
print(f"主题: {subject}")
print(f"内容: {body}")
return "邮件已发送成功!"案例3:翻译文本函数
def translate_text(text, target_language):
"""将文本翻译成目标语言。"""
# 这里会有翻译代码
return f"已将文本翻译成{target_language}: [翻译后的文本]"3. 交接(Handoffs)
交接是指一个 Agent 将对话传递给另一个更合适的 Agent。这就像一个工作人员把你转接给更专业的同事一样。
案例1:将用户转至西班牙语助手
def transfer_to_spanish_agent():
"""将用户转给西班牙语助手处理。"""
return spanish_agent案例2:将用户转至技术支持
def transfer_to_tech_support():
"""如果用户有技术问题,将其转给技术支持团队。"""
return tech_support_agent案例3:根据问题类型转接
def transfer_based_on_topic(topic):
"""根据话题将用户转给相应的专家。"""
if topic == "退款":
return refund_agent
elif topic == "订单":
return order_agent
else:
return general_agent4. 上下文变量(Context Variables)
上下文变量是在对话期间可以在不同 Agent 之间共享的信息片段。它们让 Agent 能够记住用户的信息和之前的对话内容。
案例1:记住用户名称
# 创建包含用户信息的上下文变量
context_variables = {"user_name": "小明", "user_id": 123}
# 使用这些变量运行对话
response = client.run(
agent=greeter_agent,
messages=[{"role": "user", "content": "你好!"}],
context_variables=context_variables
)
# 助手会回复:"你好,小明!很高兴见到你。"案例2:在指令中使用上下文变量
def instructions(context_variables):
name = context_variables.get("user_name", "用户")
return f"你是一个乐于助人的助手。请用用户的名字({name})来打招呼。"
personalized_agent = Agent(
name="个性化助手",
instructions=instructions
)案例3:在函数中使用上下文变量
def print_account_details(context_variables):
user_id = context_variables.get("user_id", "未知")
name = context_variables.get("user_name", "未知用户")
print(f"账户详情: 用户名={name}, ID={user_id}")
return "账户信息已显示"Swarm 工作流程(图解)
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ │ │ │ │ │
│ 用户输入 │────▶│ Swarm 客户端 │────▶│ 初始 Agent │
│ │ │ │ │ │
└─────────────────┘ └─────────────────┘ └────────┬────────┘
│
▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ │ │ │ │ │
│ 返回最终回复 │◀────│ 处理结果 │◀────│ Agent 处理请求 │
│ │ │ │ │ │
└─────────────────┘ └─────────────────┘ └────────┬────────┘
▲ │
│ ▼
│ ┌─────────────────┐
│ │ │
│ │ 需要使用函数吗? │
│ │ │
│ └────────┬────────┘
│ │
│ ▼
┌──────┴───────┐ ┌─────────────────┐
│ │ 是 │ │
│ 需要交接吗? │◀─────│ 执行函数调用 │
│ │ │ │
└──────┬───────┘ └─────────────────┘
│ 否
│
▼
┌─────────────────┐
│ │
│ Agent 提供回复 │
│ │
└─────────────────┘Swarm 代码结构
- core.py: 主要引擎,使所有功能协同工作
- types.py: 定义 Agent、Response 和 Result 的结构
- util.py: 框架使用的辅助工具
- repl: 让你可以与 Agent 聊天的工具
如何使用 Swarm(简单步骤)
步骤 1:设置环境
首先安装 Swarm(需要 Python 3.10+):
pip install git+https://github.com/openai/swarm.git步骤 2:创建一个简单的 Agent
from swarm import Swarm, Agent
# 创建 Swarm 客户端
client = Swarm()
# 创建一个简单的助手 Agent
helper = Agent(
name="我的助手",
instructions="你是一个友好的助手,总是用热情的语气回答问题。"
)步骤 3:与 Agent 交谈
# 发送消息并获取回复
response = client.run(
agent=helper,
messages=[{"role": "user", "content": "你好!"}]
)
# 查看 Agent 的回复
print(response.messages[-1]["content"])进阶功能演示
1. 使用函数功能
案例:创建一个可以查询天气的助手
from swarm import Swarm, Agent
# 创建 Swarm 客户端
client = Swarm()
# 定义天气查询函数
def get_weather(location, time="现在"):
"""获取指定地点的天气。地点必须是一个城市名。"""
return f"{location}在{time}的天气晴朗,气温25°C。"
# 创建一个带有函数的助手
weather_assistant = Agent(
name="天气助手",
instructions="你是一个天气助手,可以告诉用户任何地点的天气情况。",
functions=[get_weather]
)
# 与助手交谈
response = client.run(
agent=weather_assistant,
messages=[{"role": "user", "content": "北京今天天气怎么样?"}]
)
# 查看回复
print(response.messages[-1]["content"])2. 使用 Agent 交接功能
案例:创建可以相互交接的多语言助手
from swarm import Swarm, Agent
# 创建 Swarm 客户端
client = Swarm()
# 创建中文助手
chinese_agent = Agent(
name="中文助手",
instructions="你只说中文,用友好的语气回答问题。"
)
# 创建英文助手
english_agent = Agent(
name="English Assistant",
instructions="You only speak English. Be friendly and helpful."
)
# 创建交接函数
def transfer_to_chinese():
"""将用户转给中文助手"""
return chinese_agent
def transfer_to_english():
"""Transfer to English assistant"""
return english_agent
# 添加交接函数到助手
chinese_agent.functions.append(transfer_to_english)
english_agent.functions.append(transfer_to_chinese)
# 开始与中文助手交谈
response = client.run(
agent=chinese_agent,
messages=[{"role": "user", "content": "Can you speak English?"}]
)
# 助手会检测到英文并转接到英文助手
print(response.messages[-1]["content"])3. 使用上下文变量
案例:创建能记住用户偏好的助手
from swarm import Swarm, Agent, Result
# 创建 Swarm 客户端
client = Swarm()
# 定义保存偏好的函数
def save_preference(preference_type, value):
"""保存用户偏好"""
return Result(
value=f"已保存您的{preference_type}偏好: {value}",
context_variables={preference_type: value}
)
# 定义获取偏好的函数
def get_preference(context_variables, preference_type):
"""获取用户之前保存的偏好"""
value = context_variables.get(preference_type, "未设置")
return f"您的{preference_type}偏好是: {value}"
# 创建带有这些函数的助手
preference_agent = Agent(
name="偏好助手",
instructions="你是一个可以记住用户偏好的助手。",
functions=[save_preference, get_preference]
)
# 用户设置偏好
response1 = client.run(
agent=preference_agent,
messages=[{"role": "user", "content": "请记住我喜欢蓝色"}],
context_variables={}
)
# 获取更新后的上下文变量
updated_context = response1.context_variables
# 用户询问之前设置的偏好
response2 = client.run(
agent=preference_agent,
messages=[{"role": "user", "content": "我之前说我喜欢什么颜色?"}],
context_variables=updated_context
)
# 查看回复
print(response2.messages[-1]["content"])实用案例详解
1. 客户服务系统
from swarm import Swarm, Agent
# 创建不同类型的客服助手
triage_agent = Agent(
name="分类助手",
instructions="你的工作是确定用户的问题类型,然后转接到相应的专业助手。"
)
refund_agent = Agent(
name="退款助手",
instructions="你是退款专家,可以处理所有与退款相关的问题。"
)
order_agent = Agent(
name="订单助手",
instructions="你是订单专家,可以处理所有与订单相关的问题。"
)
# 创建转接函数
def transfer_to_refund():
"""转接到退款助手"""
return refund_agent
def transfer_to_order():
"""转接到订单助手"""
return order_agent
# 添加转接函数到分类助手
triage_agent.functions = [transfer_to_refund, transfer_to_order]
# 使用分类助手开始对话
client = Swarm()
response = client.run(
agent=triage_agent,
messages=[{"role": "user", "content": "我想申请退款,我的订单号是123456"}]
)
# 分类助手会将用户转接到退款助手
print(f"当前处理的助手: {response.agent.name}")
print(f"助手回复: {response.messages[-1]['content']}")2. 天气提醒系统
from swarm import Swarm, Agent, Result
# 天气检查函数
def check_weather(location):
"""检查指定地点的天气"""
# 假设这里有真实的天气API调用
return "晴天,温度25°C"
# 发送提醒函数
def send_reminder(context_variables, message):
"""向用户发送提醒"""
user_name = context_variables.get("user_name", "用户")
print(f"发送提醒给 {user_name}: {message}")
return f"提醒已发送"
# 保存用户位置函数
def save_location(location):
"""保存用户的位置偏好"""
return Result(
value=f"已保存您的位置: {location}",
context_variables={"user_location": location}
)
# 创建天气助手
weather_agent = Agent(
name="天气助手",
instructions="你是一个天气助手,可以检查天气并发送提醒。",
functions=[check_weather, send_reminder, save_location]
)
# 运行对话
client = Swarm()
response = client.run(
agent=weather_agent,
messages=[
{"role": "user", "content": "我住在上海,请记住这一点"},
{"role": "assistant", "content": "好的,我已记住您住在上海。"},
{"role": "user", "content": "现在天气怎么样?如果是好天气,请提醒我出去散步"}
],
context_variables={"user_name": "张先生"}
)
# 查看结果
print(response.messages[-1]["content"])3. 多语言翻译链
from swarm import Swarm, Agent
# 创建翻译函数
def translate_to_english(text):
"""将文本翻译成英文"""
# 假设这里有真实的翻译API
return f"英文翻译: {text} -> [英文翻译结果]"
def translate_to_chinese(text):
"""将文本翻译成中文"""
return f"中文翻译: {text} -> [中文翻译结果]"
def translate_to_spanish(text):
"""将文本翻译成西班牙语"""
return f"西班牙语翻译: {text} -> [西班牙语翻译结果]"
# 创建识别语言的函数
def detect_language(text):
"""识别文本的语言"""
# 简单示例逻辑
if "你好" in text or "中文" in text:
return "中文"
elif "hola" in text or "español" in text:
return "西班牙语"
else:
return "英文"
# 创建转接函数
def transfer_to_appropriate_translator(text):
"""根据检测到的语言转接到适当的翻译助手"""
language = detect_language(text.lower())
if language == "中文":
return chinese_translator
elif language == "西班牙语":
return spanish_translator
else:
return english_translator
# 创建各种翻译助手
language_detector = Agent(
name="语言识别助手",
instructions="你帮助识别用户使用的语言,并转接到相应的翻译助手。",
functions=[transfer_to_appropriate_translator]
)
english_translator = Agent(
name="英文翻译助手",
instructions="你是英文翻译专家。",
functions=[translate_to_english]
)
chinese_translator = Agent(
name="中文翻译助手",
instructions="你是中文翻译专家。",
functions=[translate_to_chinese]
)
spanish_translator = Agent(
name="西班牙语翻译助手",
instructions="你是西班牙语翻译专家。",
functions=[translate_to_spanish]
)
# 启动对话
client = Swarm()
response = client.run(
agent=language_detector,
messages=[{"role": "user", "content": "Hola, me gustaría traducir esto al chino."}]
)
# 查看结果
print(f"当前助手: {response.agent.name}")
print(response.messages[-1]["content"])测试你的 Swarm 系统
Swarm 提供了测试工具,帮助你确保你的 Agent 正常工作:
from swarm import Swarm, Agent
from tests.mock_client import MockOpenAIClient, create_mock_response
# 创建模拟客户端
mock_client = MockOpenAIClient()
mock_client.set_response(
create_mock_response({"role": "assistant", "content": "我是测试回复"})
)
# 使用模拟客户端创建 Swarm
client = Swarm(client=mock_client)
# 创建测试 Agent
test_agent = Agent(name="测试助手")
# 运行测试
response = client.run(
agent=test_agent,
messages=[{"role": "user", "content": "测试消息"}]
)
# 验证结果
assert response.messages[-1]["content"] == "我是测试回复"
print("测试通过!")实用开发技巧
- 从简单开始:先创建一个基本的 Agent,确保它能正常回答问题
- 逐步添加函数:一次添加一个函数,并测试每个函数是否正常工作
- 使用有意义的名称:给 Agent 和函数起有描述性的名称
- 清晰的指令:Agent 的指令应该非常明确,告诉它应该做什么和不应该做什么
- 使用上下文变量:利用上下文变量在多个交互中保持状态
- 测试每个路径:确保所有可能的交接路径都能正常工作
- 使用
run_demo_loop进行快速测试:from swarm.repl import run_demo_loop run_demo_loop(my_agent, stream=True)
总结:Swarm 的优势
- 轻量级:简单,不需要复杂设置
- 灵活:可以轻松更改 Agent 的行为
- 可组合:Agent 可以以多种方式协同工作
- 无状态:不需要在调用之间保存复杂的信息
Swarm 框架让你可以轻松创建由多个智能助手组成的系统,它们可以协同工作、相互交流,并使用特殊功能来帮助用户。无论你是想创建一个简单的问答机器人,还是一个复杂的多助手系统,Swarm 都提供了简单而强大的工具来实现它!